package com.itcast.bulls.stock.proxy.processor;

import com.itcast.bulls.stock.common.exception.ComponentException;
import com.itcast.bulls.stock.common.service.lock.RedisLockService;
import com.itcast.bulls.stock.common.utils.GlobalConstants;
import com.itcast.bulls.stock.proxy.netty.StockProxyServerHandler;
import com.itcast.bulls.stock.struct.grpc.trade.*;
import com.itcast.bulls.stock.struct.netty.trade.*;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import lombok.extern.log4j.Log4j2;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

@Log4j2
@Service
public class StockUserProcessor extends AbstractStockProcessor implements IBaseStockProcessor{

    /**
     * 基于Redis的分布式锁的实现接口
     */
    @Autowired
    private RedisLockService redisLockService;

    /**
     * Grpc用户服务接口
     */
    @GrpcClient("grpc-trade-server")
    private StockUserServiceGrpc.StockUserServiceBlockingStub stockUserServiceBlockingStub;


    /**
     * 负责用户接口请求数据处理
     * 代理服务对用户登录接口的处理逻辑
     * @param channelHandlerContext
     * @param stockMessage
     */
    @Override
    public void processRequest(ChannelHandlerContext channelHandlerContext, StockMessage stockMessage) {
        switch (stockMessage.getStockHead().getRequestType()) {
            case USER_LOGIN:
                userLogin(channelHandlerContext, stockMessage);
                break;
            case ACCOUNT_WARN_SETTING:
                accountWarnSetting(channelHandlerContext, stockMessage);
                break;
            default:
                log.error("Process in StockDealProcessor.processRequest method => Not Match RequestType!");
        }
    }

    void userLogin(ChannelHandlerContext channelHandlerContext, StockMessage stockMessage){
        ResponseMessage responseMessage = null;
        StockHeadProto.StockHead stockHead = null;
        try {
            //1.转换请求参数对象，获取参数信息
            LoginData loginData = stockMessage.getLoginData();
            LoginRequest loginRequest = LoginRequest.newBuilder()
                    .setUserNo(loginData.getUserNo())
                    .setUserPwd(loginData.getUserPwd())
                    .build();
            //2.调用远程的Grpc用户登录接口
            LoginResponse response = stockUserServiceBlockingStub.userLogin(loginRequest);
            if(GlobalConstants.STATUS_OK.equalsIgnoreCase(response.getStatus())){
                //3.记录用户缓存信息（用户在线缓存信息，用户账户ID和连接通道的关联缓存）
                recordCache(channelHandlerContext.channel().id(), response);
                //记录channel
                StockProxyServerHandler.channels.add(channelHandlerContext.channel());
                //4.返回成功处理报文
                stockHead = generateStockHead(stockMessage.getStockHead(),GlobalConstants.STATUS_OK);
                /*LoginDataResp loginDataResp = LoginDataResp.newBuilder()
                        .setUserId(response.getUserId())
                        .setAccountId(response.getAccountId())
                        .build();
                responseMessage = ResponseMessage.newBuilder()
                        .setStockHead(stockHead)
                        .setLoginDataResp(loginDataResp)
                        .setMessage(response.getMessage())
                        .build();*/
            }else {
                //返回失败的报文数据
                stockHead = generateStockHead(stockMessage.getStockHead(),GlobalConstants.STATUS_ERROR);
                /*responseMessage = ResponseMessage.newBuilder()
                        .setStockHead(stockHead)
                        .setMessage(response.getMessage())
                        .build();*/
            }
            responseMessage = ResponseMessage.newBuilder().setStockHead(stockHead)
                    .setMessage(response.getMessage()).build();
        }catch (Exception e){
            log.error(e.getMessage(),e);
            stockHead = generateStockHead(stockMessage.getStockHead(), GlobalConstants.STATUS_ERROR);
            responseMessage = ResponseMessage.newBuilder().setStockHead(stockHead)
                    .setMessage(e.getMessage())
                    .build();
        }
        channelHandlerContext.writeAndFlush(responseMessage);
        log.info("Process in StockUserProcessor.processRequest method,process complent");
    }


    void accountWarnSetting(ChannelHandlerContext channelHandlerContext, StockMessage stockMessage){
        ResponseMessage responseMessage = null;
        StockHeadProto.StockHead stockHead = null;
        try {
            AccountWarnSettingData warnSettingData = stockMessage.getAccountWarnSettingData();
            AccountWarnSettingRequest request = AccountWarnSettingRequest.newBuilder()
                    .setAccountId(getUserAccountId(channelHandlerContext))
                    .setStockId(warnSettingData.getStockId())
                    .setWarnType(warnSettingData.getWarnType())
                    .setStopRate(warnSettingData.getStopRate())
                    .setNotifyType(warnSettingData.getNotifyType())
                    .build();
            final AccountWarnSettingResponse response = stockUserServiceBlockingStub.accountWarnSetting(request);

            stockHead = generateStockHead(stockMessage.getStockHead(), response.getStatus());
            responseMessage = ResponseMessage.newBuilder()
                    .setStockHead(stockHead).setMessage(response.getMessage())
                    .build();
        }catch(ComponentException e) {
            // 业务异常返回
            log.error(e.getMessage(), e);
            stockHead = generateStockHead(stockMessage.getStockHead(), e.getErrorCodeEnum().getCode());
            responseMessage = ResponseMessage.newBuilder().setStockHead(stockHead).setMessage(e.getErrorCodeEnum().getMessage()).build();
        }catch (Exception e) {
            // 系统异常返回
            log.error(e.getMessage(), e);
            stockHead = generateStockHead(stockMessage.getStockHead(), GlobalConstants.STATUS_ERROR);
            responseMessage = ResponseMessage.newBuilder().setStockHead(stockHead).setMessage(e.getMessage()).build();
        }
        channelHandlerContext.writeAndFlush(responseMessage);
        log.info("Process in StockFinanceProcessor.getFinanceOverview write : " + responseMessage);
    }

    /**
     * 记录用户缓存信息，（包括用户在线缓存信息，用户账沪和连接通道的关联缓存）
     * @param channelId
     * @param response
     */
    private void recordCache(ChannelId channelId,LoginResponse response){
        //1.记录通道ID与账户ID的关联缓存信息
        Cache cache = cacheManager.getCache(GlobalConstants.STOCK_PROXY_USER_KEY);
        cache.put(channelId.asLongText(),response.getAccountId());
        //2.记录在线用户缓存
        Cache.ValueWrapper onlineWrapper = cache.get(GlobalConstants.STOCK_PROXY_USER_KEY_ONLINE);
        //3.分布式锁，防止用户同时登录，造成缓存信息记录不一致问题
        try {
            Map<Long,ChannelId> accountInfo = null;
            redisLockService.tryLock(GlobalConstants.STOCK_PROXY_USER_KEY);
            if(null!=onlineWrapper){
                //更新缓存信息
                accountInfo = (Map<Long, ChannelId>) onlineWrapper.get();
            }else {
                //新增缓存记录
                accountInfo = new HashMap<>();
            }
            accountInfo.put(response.getAccountId(), channelId);
            //更新redis的缓存
            cache.put(GlobalConstants.STOCK_PROXY_USER_KEY_ONLINE,accountInfo);
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }finally {
            //解锁处理
            redisLockService.unlock(GlobalConstants.STOCK_PROXY_USER_KEY);
        }

    }
}
